package ddz.db.redis;

import redis.clients.jedis.*;

import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

public class ClusterRedis extends AbsRedis implements IRedis {

    private int redisScanCount = 100;

    @Override
    public byte[] get(byte[] key) {
        if (key == null) {
            return null;
        }
        JedisCluster jedisCluster = JedisFactory.getFactory().getJedisCluster();
        return jedisCluster.get(key);
    }

    @Override
    public byte[] set(byte[] key, byte[] value, int expireTime) {
        if (key == null) {
            return null;
        }
        JedisCluster jedisCluster = JedisFactory.getFactory().getJedisCluster();
        jedisCluster.setex(key, expireTime, value);
        return value;
    }

    @Override
    public void del(byte[] key) {
        if (key == null) {
            return;
        }
        JedisCluster jedisCluster = JedisFactory.getFactory().getJedisCluster();
        jedisCluster.del(key);
    }

    @Override
    public Long dbSize(byte[] pattern) {
        Long dbSize = 0L;
        JedisCluster jedisCluster = JedisFactory.getFactory().getJedisCluster();
        Map<String, JedisPool> clusterNodes = jedisCluster.getClusterNodes();
        Iterator<Map.Entry<String, JedisPool>> nodeIt = clusterNodes.entrySet().iterator();
        while (nodeIt.hasNext()) {
            Map.Entry<String, JedisPool> node = nodeIt.next();
            long nodeDbSize = getDbSizeFromClusterNode(node.getValue(), pattern);
            if (nodeDbSize == 0L) {
                continue;
            }
            dbSize += nodeDbSize;
        }
        return dbSize;
    }

    @Override
    public Set<byte[]> keys(byte[] pattern) {
        Set<byte[]> keys = new HashSet<>();
        JedisCluster jedisCluster = JedisFactory.getFactory().getJedisCluster();
        Map<String, JedisPool> clusterNodes = jedisCluster.getClusterNodes();
        Iterator<Map.Entry<String, JedisPool>> nodeIt = clusterNodes.entrySet().iterator();
        while (nodeIt.hasNext()) {
            Map.Entry<String, JedisPool> node = nodeIt.next();
            Set<byte[]> nodeKeys = getKeysFromClusterNode(node.getValue(), pattern);
            if (nodeKeys == null || nodeKeys.size() == 0) {
                continue;
            }
            keys.addAll(nodeKeys);
        }

        return keys;
    }

    private Set<byte[]> getKeysFromClusterNode(JedisPool jedisPool, byte[] pattern) {
        Jedis jedis = jedisPool.getResource();
        try {
            return scanKeys(jedis, pattern);
        } finally {
            jedis.close();
        }
    }

    private long getDbSizeFromClusterNode(JedisPool jedisPool, byte[] pattern) {
        long dbSize = 0L;
        Jedis jedis = jedisPool.getResource();

        try {
            ScanParams params = new ScanParams();
            params.count(redisScanCount);
            params.match(pattern);
            byte[] cursor = ScanParams.SCAN_POINTER_START_BINARY;
            ScanResult<byte[]> scanResult;
            do {
                scanResult = jedis.scan(cursor, params);
                dbSize++;
                cursor = scanResult.getCursorAsBytes();
            } while (scanResult.getStringCursor().compareTo(ScanParams.SCAN_POINTER_START) > 0);
        } finally {
            jedis.close();
        }
        return dbSize;
    }

    public int getRedisScanCount() {
        return redisScanCount;
    }

    public void setRedisScanCount(int redisScanCount) {
        this.redisScanCount = redisScanCount;
    }
}
